package kvraft

import (
	"6.824/labgob"
	"6.824/labrpc"
	"6.824/raft"
	"bytes"
	"fmt"
	"log"
	"os"
	"sync"
	"sync/atomic"
	"time"
)

const Debug = false

func DPrintf(format string, a ...interface{}) (n int, err error) {
	if Debug {
		log.Printf(format, a...)
	}
	return
}

type Op struct {
	// Your definitions here.
	// Field names must start with capital letters,
	// otherwise RPC will break.
	//操作类型
	Type string

	Key   string
	Value string

	//客户端唯一标识
	ClientId int64
	//命令序列号
	CmdId int64
}

type KVServer struct {
	mu      sync.Mutex
	me      int
	rf      *raft.Raft
	applyCh chan raft.ApplyMsg
	dead    int32 // set by Kill()

	maxraftstate int // snapshot if log grows this big
	persister    *raft.Persister

	//当前是否正在创建快照
	snapshotInProgress int32

	// Your definitions here.
	logger *log.Logger // 日志记录器
	//本地数据库
	DB map[string]string
	//最近执行的命令的序列号
	ClientCmdSeq map[int64]int64
	// 当前index的命令对应的传输通道
	pendingChs map[int]chan *GetReply
	// 当前index的命令
	pendingOps map[int]*Op
	//已经应用的最新的日志，raft读取持久化状态可能会重新提交日志
	LastApplied int
}

type PendingOp struct {
	//执行的操作
	op Op
	//提交当前命令时的任期
	term int
	//接收结果的通道, put和append操作可以用get的覆盖
	resultChan chan GetReply
}

func (kv *KVServer) Get(args *GetArgs, reply *GetReply) {
	// Your code here.
	kv.mu.Lock()
	kv.logger.Printf("收到Get请求: ClientId=%d, CmdId=%d, Key=%s", args.ClientId%10000, args.CmdId, args.Key)

	op := Op{
		Type:     "Get",
		Key:      args.Key,
		ClientId: args.ClientId,
		CmdId:    args.CmdId,
	}

	index, term, isLeader := kv.rf.Start(op)
	if !isLeader {
		kv.mu.Unlock()
		reply.Err = ErrWrongLeader
		kv.logger.Printf("Get请求处理失败: 当前server非领导者，拒绝ClientId=%d, CmdId=%d", args.ClientId%10000, args.CmdId)
		return
	}
	//用有缓冲通道来避免出现无人接收的极端情况，减少一个goroutine的浪费
	ch := make(chan *GetReply, 1)
	kv.pendingChs[index] = ch
	kv.pendingOps[index] = &op
	kv.logger.Printf("Get命令已提交到Raft: index=%d, term=%d, ClientId=%d, CmdId=%d", index, term, args.ClientId%10000, args.CmdId)
	kv.mu.Unlock()

	//最多等待时长
	timer := time.NewTimer(500 * time.Millisecond)
	defer timer.Stop()

	for {
		select {
		//在超时之前收到了命令
		case result := <-ch:
			// 清理映射表
			kv.mu.Lock()
			delete(kv.pendingChs, index)
			delete(kv.pendingOps, index)
			kv.mu.Unlock()

			reply.Value, reply.Err = result.Value, reply.Err
			kv.logger.Printf("Get请求处理完成: ClientId=%d, CmdId=%d, Key=%s, Value=%s, Err=%v",
				args.ClientId%10000, args.CmdId, args.Key, result.Value, result.Err)
			return
		//等待超时：
		case <-timer.C:
			// 清理映射表
			kv.mu.Lock()
			delete(kv.pendingChs, index)
			delete(kv.pendingOps, index)
			kv.mu.Unlock()

			//可能当前节点是旧的领导者，处于少数分区，需要重新寻找新的领导者
			reply.Err = ErrWrongLeader
			kv.logger.Printf("Get请求超时: ClientId=%d, CmdId=%d, index=%d", args.ClientId%10000, args.CmdId, index)
			return
		//没一百毫秒检查一下是否还是领导者且任期没有变化，任期变化即使是领导者，当前命令可能也已经不见了
		case <-time.After(100 * time.Millisecond):
			if kv.rf.StateChanged(term) {
				// 清理映射表
				kv.mu.Lock()
				delete(kv.pendingChs, index)
				delete(kv.pendingOps, index)
				kv.mu.Unlock()

				reply.Err = ErrWrongLeader
				kv.logger.Printf("Get请求处理失败: 领导者状态变更，ClientId=%d, CmdId=%d", args.ClientId%10000, args.CmdId)
				return
			}
		}
	}

}

func (kv *KVServer) PutAppend(args *PutAppendArgs, reply *PutAppendReply) {
	// Your code here.
	kv.mu.Lock()
	kv.logger.Printf("收到%s请求: ClientId=%d, CmdId=%d, Key=%s, Value=%s",
		args.Op, args.ClientId%10000, args.CmdId, args.Key, args.Value)
	op := Op{
		Type:     args.Op,
		Key:      args.Key,
		Value:    args.Value,
		ClientId: args.ClientId,
		CmdId:    args.CmdId,
	}
	index, term, isLeader := kv.rf.Start(op)
	if !isLeader {
		kv.mu.Unlock()
		reply.Err = ErrWrongLeader
		kv.logger.Printf("%s请求处理失败: 非领导者，ClientId=%d, CmdId=%d", args.Op, args.ClientId%10000, args.CmdId)
		return
	}

	// 创建通道并记录到映射表
	ch := make(chan *GetReply, 1)
	kv.pendingChs[index] = ch
	kv.pendingOps[index] = &op
	kv.logger.Printf("%s命令已提交到Raft: index=%d, term=%d, ClientId=%d, CmdId=%d",
		args.Op, index, term, args.ClientId%10000, args.CmdId)
	kv.mu.Unlock()

	// 最多等待时长
	timer := time.NewTimer(500 * time.Millisecond)
	defer timer.Stop()

	for {
		select {
		// 在超时之前收到了命令
		case result := <-ch:
			// 清理映射表
			kv.mu.Lock()
			delete(kv.pendingChs, index)
			delete(kv.pendingOps, index)
			kv.mu.Unlock()

			reply.Err = result.Err
			kv.logger.Printf("%s请求处理完成: ClientId=%d, CmdId=%d, Err=%v",
				args.Op, args.ClientId%10000, args.CmdId, result.Err)
			return

		// 等待超时
		case <-timer.C:
			// 清理映射表
			kv.mu.Lock()
			delete(kv.pendingChs, index)
			delete(kv.pendingOps, index)
			kv.mu.Unlock()

			reply.Err = ErrWrongLeader
			kv.logger.Printf("%s请求超时: ClientId=%d, CmdId=%d, index=%d", args.Op, args.ClientId%10000, args.CmdId, index)
			return

		// 每一百毫秒检查一下是否还是领导者且任期没有变化
		case <-time.After(100 * time.Millisecond):
			if kv.rf.StateChanged(term) {
				// 清理映射表
				kv.mu.Lock()
				delete(kv.pendingChs, index)
				delete(kv.pendingOps, index)
				kv.mu.Unlock()

				reply.Err = ErrWrongLeader
				kv.logger.Printf("%s请求处理失败: 领导者状态变更，ClientId=%d, CmdId=%d", args.Op, args.ClientId%10000, args.CmdId)
				return
			}
		}
	}
}

func (kv *KVServer) checkDuplicated(op *Op) bool {
	id, ok := kv.ClientCmdSeq[op.ClientId]
	//检查是否存在客户端对应信息已经是否已经过期
	return ok && id >= op.CmdId
}

func (kv *KVServer) doCmd(op *Op, index int) {
	kv.logger.Printf("开始执行命令: index=%d, Type=%s, ClientId=%d, CmdId=%d, Key=%s",
		index, op.Type, op.ClientId%10000, op.CmdId, op.Key)

	reply := new(GetReply)

	if op.Type == "Get" { //get命令无条件返回，查多少次都返回，因为是幂等的
		val, ok := kv.DB[op.Key]
		if ok {
			reply.Value, reply.Err = val, OK
			kv.logger.Printf("Get命令执行成功: index=%d, Key=%s, Value=%s", index, op.Key, val)
		} else {
			reply.Value, reply.Err = "", ErrNoKey
			kv.logger.Printf("Get命令执行成功: index=%d, Key=%s（不存在）", index, op.Key)
		}
	} else if kv.checkDuplicated(op) { //命令已经处理过了
		reply.Err = ErrDuplicate
		kv.logger.Printf("重复命令跳过执行: index=%d, Type=%s, ClientId=%d, CmdId=%d",
			index, op.Type, op.ClientId%10000, op.CmdId)
	} else { //执行命令
		if op.Type == "Put" {
			kv.DB[op.Key] = op.Value
			reply.Err = OK
			kv.logger.Printf("Put命令执行成功: index=%d, Key=%s, Value=%s", index, op.Key, op.Value)
		} else {
			val, ok := kv.DB[op.Key]
			//key存在，追加
			if ok {
				kv.DB[op.Key] = val + op.Value
			} else {
				kv.DB[op.Key] = op.Value
			}
			//kv.logger.Printf("Append命令执行成功: index=%d, Key=%s, 追加后Value=%s",
			//	index, op.Key, kv.db[op.Key])
			kv.logger.Printf("Append命令执行成功: index=%d, Key=%s",
				index, op.Key)
			reply.Err = OK
		}
		//增加当前已处理的最大命令序列号
		oldSeq := kv.ClientCmdSeq[op.ClientId] // 记录旧序列号（默认0如果是新客户端）
		kv.ClientCmdSeq[op.ClientId] = op.CmdId
		kv.logger.Printf("已更新客户端命令序列号: ClientId=%d, 旧序列号=%d → 新序列号=%d",
			op.ClientId%10000, oldSeq, op.CmdId)
	}

	//检查是否需要回复客户端
	ch, chExists := kv.pendingChs[index]
	oldOp, opExists := kv.pendingOps[index]

	// 验证通道存在且命令内容匹配
	if chExists && opExists &&
		oldOp.ClientId == op.ClientId &&
		oldOp.CmdId == op.CmdId {
		kv.logger.Printf("发送命令结果到通道: index=%d, ClientId=%d, CmdId=%d",
			index, op.ClientId%10000, op.CmdId)
		go func() { ch <- reply }()
	} else {
		kv.logger.Printf("无等待通道接收结果: index=%d（可能已超时或领导者变更或非接受命令的server）", index)
	}
}

func (kv *KVServer) createSnapshot() []byte {
	w := new(bytes.Buffer)
	e := labgob.NewEncoder(w)
	e.Encode(kv.DB)
	e.Encode(kv.ClientCmdSeq)
	e.Encode(kv.LastApplied)
	data := w.Bytes()
	return data
}

func (kv *KVServer) ReadAndApplySnapshot(data []byte) {
	if data == nil || len(data) < 1 { // bootstrap without any state?
		return
	}
	r := bytes.NewBuffer(data)
	d := labgob.NewDecoder(r)

	var DB map[string]string
	var ClientCmdSeq map[int64]int64
	var LastApplied int

	if d.Decode(&DB) != nil || d.Decode(&ClientCmdSeq) != nil || d.Decode(&LastApplied) != nil {
		kv.logger.Fatalf("Decode Error\n")
	}

	kv.mu.Lock()
	defer kv.mu.Unlock()

	//可能因为并发问题，导致快照过期了
	if kv.LastApplied >= LastApplied {
		kv.logger.Printf("KVServer %d: ignore snapshot (index %d) since lastApplied is %d",
			kv.me, LastApplied, kv.LastApplied)
		return
	}
	kv.DB = DB
	kv.ClientCmdSeq = ClientCmdSeq
	kv.LastApplied = LastApplied
}

//循环接收raft提交的命令
func (kv *KVServer) applier() {
	for !kv.killed() {
		msg := <-kv.applyCh
		kv.logger.Printf("收到raft提交的请求")
		//命令
		if msg.CommandValid {
			//类型断言并赋值
			index := msg.CommandIndex
			kv.mu.Lock()
			kv.logger.Printf("收到Raft提交的命令: index=%d, term=%d", index, msg.CommandTerm)
			//只有新日志才处理
			if index > kv.LastApplied {
				//更新已提交日志索引
				oldLastApplied := kv.LastApplied
				kv.LastApplied = index
				kv.logger.Printf("已更新已应用日志索引: 旧索引=%d → 新索引=%d", oldLastApplied, index)
				//处理任务
				op := msg.Command.(Op)
				kv.doCmd(&op, index)

				//需要快照，且raft持久化信息过大
				if kv.maxraftstate != -1 && kv.persister.RaftStateSize() >= kv.maxraftstate && atomic.LoadInt32(&kv.snapshotInProgress) == 0 {
					//标记正在创建快照
					atomic.StoreInt32(&kv.snapshotInProgress, 1)
					currentSize := kv.persister.RaftStateSize()
					threshold := kv.maxraftstate
					kv.logger.Printf("当前raft节点持久化数据超过阈值，创建并更新节点快照信息 [当前大小: %d 字节, 阈值: %d 字节]", currentSize, threshold)
					//根据自身状态创建快照
					data := kv.createSnapshot()
					//让节点更新快照数据,后台执行，不用担心顺序问题，因为raft节点执行时会判断需要保存快照
					go func(index int) {
						defer atomic.StoreInt32(&kv.snapshotInProgress, 0)
						kv.rf.Snapshot(index, data)

					}(index)
				}
			} else {
				kv.logger.Printf("命令已处理过: index=%d（当前lastApplied=%d）", index, kv.LastApplied)
			}
			kv.mu.Unlock()
		} else if msg.SnapshotValid { //快照类型的数据
			kv.ReadAndApplySnapshot(msg.Snapshot)
		}
	}
}

//
// the tester calls Kill() when a KVServer instance won't
// be needed again. for your convenience, we supply
// code to set rf.dead (without needing a lock),
// and a killed() method to test rf.dead in
// long-running loops. you can also add your own
// code to Kill(). you're not required to do anything
// about this, but it may be convenient (for example)
// to suppress debug output from a Kill()ed instance.
//
func (kv *KVServer) Kill() {
	atomic.StoreInt32(&kv.dead, 1)
	kv.rf.Kill()
	// Your code here, if desired.
	kv.logger.Println("服务器已停止")
}

func (kv *KVServer) killed() bool {
	z := atomic.LoadInt32(&kv.dead)
	return z == 1
}

//
// servers[] contains the ports of the set of
// servers that will cooperate via Raft to
// form the fault-tolerant key/value service.
// me is the index of the current server in servers[].
// the k/v server should store snapshots through the underlying Raft
// implementation, which should call persister.SaveStateAndSnapshot() to
// atomically save the Raft state along with the snapshot.
// the k/v server should snapshot when Raft's saved state exceeds maxraftstate bytes,
// in order to allow Raft to garbage-collect its log. if maxraftstate is -1,
// you don't need to snapshot.
// StartKVServer() must return quickly, so it should start goroutines
// for any long-running work.
//
func StartKVServer(servers []*labrpc.ClientEnd, me int, persister *raft.Persister, maxraftstate int) *KVServer {
	// call labgob.Register on structures you want
	// Go's RPC library to marshall/unmarshall.
	labgob.Register(Op{})

	kv := new(KVServer)
	kv.me = me
	kv.maxraftstate = maxraftstate

	// You may need initialization code here.

	kv.applyCh = make(chan raft.ApplyMsg)
	kv.rf = raft.Make(servers, me, persister, kv.applyCh)

	// You may need initialization code here.
	kv.DB = make(map[string]string)
	kv.ClientCmdSeq = make(map[int64]int64)
	kv.logger = log.New(os.Stdout, fmt.Sprintf("server[%d]: ", me), log.LstdFlags|log.Lmicroseconds)
	kv.pendingChs = make(map[int]chan *GetReply)
	kv.pendingOps = make(map[int]*Op)
	kv.persister = persister

	kv.ReadAndApplySnapshot(persister.ReadSnapshot())

	go kv.applier()

	return kv
}
